package main

import (
	"fmt"
	"influxdbDemoData/influxdb"
	"influxdbDemoData/mysql"
	"influxdbDemoData/sysinit"
	"influxdbDemoData/util"
	"github.com/astaxie/beego"
	"github.com/robfig/cron"
	"log"
	"strconv"
	"time"
)

var influxdbConnectMysqlInfluxdb *influxdb.Connect
var timeExecMysqlInfluxdb TimeExecMysqlInfluxdb

type TimeExecMysqlInfluxdb struct {
	Status    string
	StartTime string
	EndTime   string
	Cron      string
}

func main() {
	log.Println("====== main start ======")
	ch := make(chan int, 1)
	initMysqlInfluxdb()
	RunFuncMysqlInfluxdb()
	if timeExecMysqlInfluxdb.Status == "true" {
		for _, date := range util.ComputeTime(timeExecMysqlInfluxdb.StartTime, timeExecMysqlInfluxdb.EndTime) {
			execMysqlInfluxdb(date)
		}
	}
	for {
		<-ch
	}
	log.Println("====== main exit ======")

}

func RunFuncMysqlInfluxdb() {
	c := cron.New()
	c.AddFunc(timeExecMysqlInfluxdb.Cron, func() {
		h, _ := time.ParseDuration("-3h")
		now := time.Now().Add(h)
		log.Println(fmt.Sprintf("---- %v", now))
		today := now.Format(util.LayoutDate)
		log.Println(fmt.Sprintf("today %v", today))
		execMysqlInfluxdb(today)
	})
	c.Start()
	log.Println(fmt.Sprintf("cron %+v", timeExecMysqlInfluxdb.Cron))
}

func execMysqlInfluxdb(date string) {
	flowMysqlInfluxdb(date)
	packFlowMysqlInfluxdb(date)
}

func flowMysqlInfluxdb(date string) {
	pf, err := mysql.SelectPassengerflowByDate(date)
	if err != nil {
		panic(err)
	}
	var flows []influxdb.Flow

	for _, value := range *pf {
		flow := influxdb.Flow{
			Time:       value.Time,
			ZoneId:     strconv.Itoa(value.ZoneId),
			CountEnter: value.CountEnter,
			CountExit:  value.CountExit,
		}
		flows = append(flows, flow)
	}
	log.Println(fmt.Sprintf("date:%v find: %v", date, len(*pf)))
	log.Println(fmt.Sprintf("-------- flows: %v", len(flows)))
	//log.Println(fmt.Sprintf("-------- flows: %+v", flows))
	insert := influxdbConnectMysqlInfluxdb.InsertPassengerflow(&flows, influxdb.Passengerflow)
	log.Println(fmt.Sprintf("-------- insert flows: %v", insert))
}

func packFlowMysqlInfluxdb(date string) {
	p5f, err := mysql.SelectPack5PassengerflowByDate(date)
	if err != nil {
		panic(err)
	}
	var flows []influxdb.Flow

	for _, value := range *p5f {
		var date, err = time.Parse(util.LayoutDate, value.Date)
		if err != nil {
			panic(err)
		}
		var time = time.Date(date.Year(), date.Month(), date.Day(), value.DateHour, value.DateMin, 0, 0, time.Local)
		flow := influxdb.Flow{
			Time:       time,
			ZoneId:     strconv.Itoa(value.ZoneId),
			CountEnter: value.CountEnter,
			CountExit:  value.CountExit,
		}
		flows = append(flows, flow)
	}
	log.Println(fmt.Sprintf("date:%v find: %v", date, len(*p5f)))
	log.Println(fmt.Sprintf("-------- pack5f: %v", len(flows)))

	bussiness := influxdbConnectMysqlInfluxdb.InsertPassengerflow(&flows, influxdb.PassengerflowBusiensss)
	log.Println(fmt.Sprintf("-------- insert PassengerflowBusiensss: %v", bussiness))

	influxdbConnectMysqlInfluxdb.CopyInsertInto(influxdb.PassengerflowBusiensss, influxdb.Pack5passengerflow, fmt.Sprintf("%v 00:00:00", date), fmt.Sprintf("%v 23:59:59", date), "5m")
	influxdbConnectMysqlInfluxdb.CopyInsertInto(influxdb.Pack5passengerflow, influxdb.Pack60Passengerflow, fmt.Sprintf("%v 00:00:00", date), fmt.Sprintf("%v 23:59:59", date), "60m")
}

func initMysqlInfluxdb() {
	beego.LoadAppConfig("ini", "conf/app-mysqlinfluxdb.conf")
	sysinit.InitDatabase([]string{"mysql"})
	sysinit.InitLogs()

	//创建链接Influxdb
	iaddr := beego.AppConfig.String("influxdb::addr")
	ipassword := beego.AppConfig.String("influxdb::password")
	iusername := beego.AppConfig.String("influxdb::username")
	idbname := beego.AppConfig.String("influxdb::dbname")
	itimezone := beego.AppConfig.String("influxdb::timezone")

	influxdbConnectMysqlInfluxdb = influxdb.NewConnect(iaddr, iusername, ipassword, idbname, itimezone)

	//TimeInterval
	timeExecMysqlInfluxdb.Status = beego.AppConfig.String("timeInterval::status")
	timeExecMysqlInfluxdb.StartTime = beego.AppConfig.String("timeInterval::start_time")
	timeExecMysqlInfluxdb.EndTime = beego.AppConfig.String("timeInterval::end_time")
	timeExecMysqlInfluxdb.Cron = beego.AppConfig.String("timeInterval::cron")
}
